package com.hefei.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.junit.Test;

import java.io.File;
import java.io.FileInputStream;
import java.net.URI;

/**
 * @author caizhengjun
 * @date 2020/8/26 8:24 下午
 */
public class HdfsClient {

    //JVM可指定启动参数 -DHADOOP_USER_NAME=username

    //创建目录
    @Test
    public void test() throws Exception {

        //1 获取文件系统
        Configuration configuration = new Configuration();

        //配置在集群运行
        configuration.set("fs.defaultFS", "hdfs://work01:9000");

        URI uri = new URI("hdfs://work01:9000");
        FileSystem fs = FileSystem.get(uri, configuration, "root");

        //2 创建目录
        Path path = new Path("2020", "08-26");
        fs.mkdirs(path);

        fs.close();
    }

    //文件上传
    @Test
    public void upload() throws Exception {
        Configuration configuration = new Configuration();
        configuration.set("dfs.replication", "3");
        URI uri = new URI("hdfs://work01:9000");
        FileSystem fs = FileSystem.get(uri, configuration, "root");
        Path srcPath = new Path("/Users/caizhengjun/wordcount.txt");
        Path dstPath = new Path("/user/caizhengjun/input");
        fs.copyFromLocalFile(srcPath, dstPath);

        fs.close();
        System.out.println("文件上传结束");
    }

    //文件下载
    @Test
    public void download() throws Exception {
        Configuration configuration = new Configuration();
        URI uri = new URI("hdfs://work01:9000");
        FileSystem fs = FileSystem.get(uri, configuration, "root");
        Path srcPath = new Path("/user/caizhengjun/input/wordcount.txt");
        Path dstPath = new Path("/Users/caizhengjun/wordcount888.txt");
        fs.copyToLocalFile(srcPath, dstPath);
        fs.close();
    }

    //文件夹删除
    @Test
    public void deleteDir() throws Exception {
        Configuration configuration = new Configuration();
        URI uri = new URI("hdfs://work01:9000");
        FileSystem fs = FileSystem.get(uri, configuration, "root");
        Path path = new Path("/user/caizhengjun/input/wc.txt");
        fs.deleteOnExit(path);
        fs.close();
    }

    //文件名更改
    @Test
    public void rename() throws Exception {
        Configuration configuration = new Configuration();
        URI uri = new URI("hdfs://work01:9000");
        FileSystem fs = FileSystem.get(uri, configuration, "root");
        Path srcPath = new Path("/user/caizhengjun/input/wordcount.txt");
        Path dstPath = new Path("/user/caizhengjun/input/wordcount888.txt");
        fs.rename(srcPath, dstPath);

        fs.close();
    }

    @Test
    public void streamUpload() throws Exception {
        Configuration configuration = new Configuration();
        URI uri = new URI("hdfs://work01:9000");
        FileSystem fs = FileSystem.get(uri, configuration, "root");
        FileInputStream is = new FileInputStream(new File("/Users/caizhengjun/wordcount.txt"));
        FSDataOutputStream os = fs.create(new Path("/user/caizhengjun/input/wordcountStreamCopy.txt"));

        IOUtils.copyBytes(is, os, configuration);

        IOUtils.closeStream(os);
        IOUtils.closeStream(is);
        fs.close();
    }
}